package org.apache.rocketmq.broker;

import com.google.common.collect.Lists;
import org.apache.rocketmq.broker.client.*;
import org.apache.rocketmq.broker.latency.BrokerFastFailure;
import org.apache.rocketmq.broker.longpolling.NotifyMessageArrivingListener;
import org.apache.rocketmq.broker.mqtrace.SendMessageHook;
import org.apache.rocketmq.broker.out.BrokerOuterAPI;
import org.apache.rocketmq.broker.plugin.BrokerAttachedPlugin;
import org.apache.rocketmq.broker.processor.AdminBrokerProcessor;
import org.apache.rocketmq.broker.processor.SendMessageProcessor;
import org.apache.rocketmq.broker.topic.TopicConfigManager;
import org.apache.rocketmq.broker.topic.TopicQueueMappingManager;
import org.apache.rocketmq.broker.util.HookUtils;
import org.apache.rocketmq.common.*;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.constant.PermName;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.utils.ServiceProvider;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.Configuration;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.RemotingServer;
import org.apache.rocketmq.remoting.common.TlsMode;
import org.apache.rocketmq.remoting.netty.*;
import org.apache.rocketmq.remoting.protocol.DataVersion;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.body.TopicConfigAndMappingSerializeWrapper;
import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
import org.apache.rocketmq.remoting.protocol.namesrv.RegisterBrokerResult;
import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingDetail;
import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingInfo;
import org.apache.rocketmq.srvutil.FileWatchService;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.MessageArrivingListener;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.PutMessageResult;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.hook.PutMessageHook;
import org.apache.rocketmq.store.hook.SendMessageBackHook;
import org.apache.rocketmq.store.plugin.MessageStoreFactory;
import org.apache.rocketmq.store.plugin.MessageStorePluginContext;
import org.apache.rocketmq.store.stats.BrokerStats;
import org.apache.rocketmq.store.stats.BrokerStatsManager;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import java.util.stream.Collectors;


/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/1/9
 * @方法描述：Broker模块的控制器，该控制器负责管理Broker的各个组件和服务，包括网络通信、定时任务、配置管理等。
 */
public class BrokerController {


    protected static final Logger LOG = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);

    private static final Logger LOG_PROTECTION = LoggerFactory.getLogger(LoggerName.PROTECTION_LOGGER_NAME);

    private static final Logger LOG_WATER_MARK = LoggerFactory.getLogger(LoggerName.WATER_MARK_LOGGER_NAME);

    //Broker模块的配置信息对象
    protected final BrokerConfig brokerConfig;
    //Netty服务端的配置信息对象
    private final NettyServerConfig nettyServerConfig;
    //Netty客户端的配置信息对象
    private final NettyClientConfig nettyClientConfig;
    //客户端连接清理器，主要用来负责扫描并清理失效的客户端连接，在第五版本代码中还用不到这个客户端连接清理器
    protected final ClientHousekeepingService clientHousekeepingService;
    //Broker对外发送请求的API组件
    protected BrokerOuterAPI brokerOuterAPI;
    //定时任务执行器
    protected ScheduledExecutorService scheduledExecutorService;
    //专门用来处理控制台向Broker发送的请求的，但在第五版本代码中，这个执行器还用不到，第六版本就会用到了
    protected ExecutorService adminBrokerExecutor;
    //这个任务队列也是，在第五版本还用不到，第六版本就会用到了
    protected final BlockingQueue<Runnable> adminBrokerThreadPoolQueue;
    //Broker的Netty服务端
    protected RemotingServer remotingServer;
    //快速响应的Netty服务端，当remotingServer服务端处理不过来请求的时候，就会交给这个快速响应的Netty服务端来处理
    protected RemotingServer fastRemotingServer;
    //Broker模块的配置管理器
    private Configuration configuration;
    //当前Broker程序是否关闭的标志
    protected volatile boolean shutdown = false;
    //Broker程序关闭时要被回调的钩子方法
    protected ShutdownHook shutdownHook;
    //存储已提交的定时任务的Future集合
    protected final List<ScheduledFuture<?>> scheduledFutures = new ArrayList<>();
    //判断当前Broker模块是否处于隔离状态的标志，处于隔离状态的Broker模块不会向NameServer注册信息
    //这个标志在第五版本代码中还用不到，引入集群之后才会用到，所以现在它总是false
    protected volatile boolean isIsolated = false;
    //存储Broker插件的集合，用于扩展Broker的功能
    protected List<BrokerAttachedPlugin> brokerAttachedPlugins = new ArrayList<>();
    //Broker程序的统计管理器，用于收集和管理Broker的运行统计数据
    protected final BrokerStatsManager brokerStatsManager;
    //Broker主题信息配置管理器，所有在Broker创建的主题信息，都会交给这个管理器来管理
    protected TopicConfigManager topicConfigManager;
    //主题队列映射管理器，管理Broker模块的所有主题队列映射信息
    protected TopicQueueMappingManager topicQueueMappingManager;
    //生产者信息管理器
    protected final ProducerManager producerManager;

    protected final ConsumerIdsChangeListener consumerIdsChangeListener;
    //消费者信息管理器
    protected final ConsumerManager consumerManager;
    //处理生产者发送过来的消息的请求处理器
    protected final SendMessageProcessor sendMessageProcessor;
    //发送消息前后的回调方法集合，用于在处理发送消息请求的过程中插入自定义逻辑
    protected final List<SendMessageHook> sendMessageHookList = new ArrayList<>();

    //该成员变量在第十三版本代码中还用不到，我怕之后忘了这个成员变量，所以先写在这里
    //protected final List<ConsumeMessageHook> consumeMessageHookList = new ArrayList<>();

    //负责处理发送消息请求的执行器
    protected ExecutorService sendMessageExecutor;
    //处理发送消息请求的执行器的任务队列
    protected final BlockingQueue<Runnable> sendThreadPoolQueue;
    //消息存储引擎，负责消息的持久化存储
    protected MessageStore messageStore;
    //消息到达监听器，用于监听消息到达事件并触发相应的处理
    protected final MessageArrivingListener messageArrivingListener;
    //存储引擎配置信息类
    protected final MessageStoreConfig messageStoreConfig;

    //Broker信息统计器，负责收集和报告Broker的运行统计信息
    private BrokerStats brokerStats;

    private InetSocketAddress storeHost;

    protected ExecutorService putMessageFutureExecutor;

    protected final BlockingQueue<Runnable> putThreadPoolQueue;


    //构造方法
    public BrokerController(
            final BrokerConfig brokerConfig,
            final NettyServerConfig nettyServerConfig,
            final NettyClientConfig nettyClientConfig,
            MessageStoreConfig messageStoreConfig,
            final ShutdownHook shutdownHook
    ) {
        this(brokerConfig, nettyServerConfig, nettyClientConfig,messageStoreConfig);
        this.shutdownHook = shutdownHook;
    }


    //构造方法
    public BrokerController(final BrokerConfig brokerConfig, final NettyServerConfig nettyServerConfig, final NettyClientConfig nettyClientConfig,MessageStoreConfig messageStoreConfig) {

        this.brokerConfig = brokerConfig;
        this.nettyServerConfig = nettyServerConfig;
        this.nettyClientConfig = nettyClientConfig;
        this.messageStoreConfig = messageStoreConfig;
        this.setStoreHost(new InetSocketAddress(this.getBrokerConfig().getBrokerIP1(), getListenPort()));
        this.brokerStatsManager = new BrokerStatsManager(this.brokerConfig.getBrokerClusterName(), this.brokerConfig.isEnableDetailStat());
        //在这里创建主题配置管理器，管理Broker模块的所有主题配置信息
        this.topicConfigManager = new TopicConfigManager(this);
        //在这里创建主题队列映射管理器，管理Broker模块的所有主题队列映射信息
        this.topicQueueMappingManager = new TopicQueueMappingManager(this);
        if (nettyClientConfig != null) {
            //创建Broker对外发送请求的API组件时，Broker模块的Netty客户端也会构建完毕
            this.brokerOuterAPI = new BrokerOuterAPI(nettyClientConfig);
        }
        this.clientHousekeepingService = new ClientHousekeepingService(this);
        this.adminBrokerThreadPoolQueue = new LinkedBlockingQueue<>(this.brokerConfig.getAdminBrokerThreadPoolQueueCapacity());
        this.sendThreadPoolQueue = new LinkedBlockingQueue<>(this.brokerConfig.getSendThreadPoolQueueCapacity());
        this.putThreadPoolQueue = new LinkedBlockingQueue<>(this.brokerConfig.getPutThreadPoolQueueCapacity());
        this.consumerIdsChangeListener = new DefaultConsumerIdsChangeListener(this);
        this.consumerManager = new ConsumerManager(this.consumerIdsChangeListener, this.brokerStatsManager, this.brokerConfig);
        this.sendMessageProcessor = new SendMessageProcessor(this);
        this.producerManager = new ProducerManager(this.brokerStatsManager);
        this.messageArrivingListener = new NotifyMessageArrivingListener();

        //定义Broker配置文件的路径
        String brokerConfigPath;
        //判断Broker配置信息对象中是否设置了Broker配置文件的路径
        if (brokerConfig.getBrokerConfigPath() != null && !brokerConfig.getBrokerConfigPath().isEmpty()) {
            //如果设置了，则使用设置的路径
            brokerConfigPath = brokerConfig.getBrokerConfigPath();
        } else {
            //如果没有设置，则使用默认的路径
            //在BrokerStartup类中，已经把配置文件路径设置到了BrokerPathConfigHelper中
            brokerConfigPath = BrokerPathConfigHelper.getBrokerConfigPath();
        }
        //创建Broker的配置管理器，管理Broker模块要用到的所有配置信息
        this.configuration = new Configuration(LOG, brokerConfigPath, this.brokerConfig, this.nettyServerConfig, this.nettyClientConfig);
    }



    //初始化Broker的方法
    public boolean initialize() throws CloneNotSupportedException {

        //初始化元数据，在第七版本代码中，这里初始化元数据的操作
        //仅仅是把Broker存储得主题配置信息从本地加载到内存中
        boolean result = this.initializeMetadata();
        //元数据初始化失败
        if (!result) {
            //直接返回false
            return false;
        }
        //初始化消息存储组件
        result = this.initializeMessageStore();
        if (!result) {
            return false;
        }
        return this.recoverAndInitService();
    }


    //初始化Broker元数据的方法
    public boolean initializeMetadata() {
        //加载主题配置信息
        boolean result = this.topicConfigManager.load();
        //返回初始化结果
        return result;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：初始化消息存储组件的方法
     */
    public boolean initializeMessageStore() {
        //初始化结果标志，默认为true
        boolean result = true;
        try {
            //创建默认的消息存储引擎，源码中这里其实还有一个判断，那就是是否启用了RocksDBMessageStore存储引擎，没有启用则创建默认的消息存储引擎
            //因为源码中使用的就是默认的，所以我这里就直接省去了判断，直接创建默认的消息存储引擎了
            DefaultMessageStore defaultMessageStore = new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener, this.brokerConfig, topicConfigManager.getTopicConfigTable());
            //创建Broker信息统计对象
            this.brokerStats = new BrokerStats(defaultMessageStore);
            //创建消息存储引擎插件对象
            MessageStorePluginContext context = new MessageStorePluginContext(messageStoreConfig, brokerStatsManager, messageArrivingListener, brokerConfig, configuration);
            //使用工厂对象创建消息存储引擎
            this.messageStore = MessageStoreFactory.build(context, defaultMessageStore);
        } catch (IOException e) {
            result = false;
            LOG.error("BrokerController#initialize: unexpected error occurs", e);
        }
        return result;
    }


    //初始化Broker各个服务组件的方法
    public boolean recoverAndInitService() throws CloneNotSupportedException {
        boolean result = true;
        //如果消息存储引擎不为空
        if (messageStore != null) {
            //注册消息存储引擎的回调方法
            registerMessageStoreHook();
            //加载消息存储引擎的数据
            result = this.messageStore.load();
        }
        //遍历加载Broker的各个插件，加载成功则返回true，否则返回false
        for (BrokerAttachedPlugin brokerAttachedPlugin : brokerAttachedPlugins) {
            if (brokerAttachedPlugin != null) {
                result = result && brokerAttachedPlugin.load();
            }
        }
        //初始化各个服务组件
        if (result) {
            //初始化远程通信组件
            initializeRemotingServer();
            //初始化线程池资源
            initializeResources();
            //注册请求处理器
            registerProcessor();
            //初始化各个定时任务
            initializeScheduledTasks();
            //初始化RPC回调方法
            initialRpcHooks();
        }
        return result;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：注册消息存储引擎回调方法的操作
     */
    public void registerMessageStoreHook() {
        List<PutMessageHook> putMessageHookList = messageStore.getPutMessageHookList();

        putMessageHookList.add(new PutMessageHook() {
            @Override
            public String hookName() {
                return "checkBeforePutMessage";
            }

            @Override
            public PutMessageResult executeBeforePutMessage(MessageExt msg) {
                //执行存储消息前的检查操作
                return HookUtils.checkBeforePutMessage(BrokerController.this, msg);
            }
        });


        //创建一个发送消息回调的钩子
        SendMessageBackHook sendMessageBackHook = new SendMessageBackHook() {
            @Override
            public boolean executeSendMessageBack(List<MessageExt> msgList, String brokerName, String brokerAddr) {
                //执行发送消息回调
                return HookUtils.sendMessageBack(BrokerController.this, msgList, brokerName, brokerAddr);
            }
        };

        //如果消息存储组件不为空，则设置发送消息回调钩子
        if (messageStore != null) {
            messageStore.setSendMessageBackHook(sendMessageBackHook);
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/9
     * @方法描述：初始化远程通信组件的方法
     */
    protected void initializeRemotingServer() throws CloneNotSupportedException {
        //创建Netty服务器
        this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, clientHousekeepingService);
        //创建快速响应服务器的配置信息
        NettyServerConfig fastConfig = (NettyServerConfig) this.nettyServerConfig.clone();
        //得到快速响应服务器要坚挺的端口号
        int listeningPort = nettyServerConfig.getListenPort() - 2;
        if (listeningPort < 0) {
            listeningPort = 0;
        }
        //设置快速响应服务器要坚挺的端口号
        fastConfig.setListenPort(listeningPort);
        //创建快速响应服务器
        this.fastRemotingServer = new NettyRemotingServer(fastConfig, clientHousekeepingService);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/9
     * @方法描述：初始化线程池资源的方法
     */
    protected void initializeResources() {
        //创建定时任务执行器，getBrokerIdentity()方法得到的就是当前Broker节点的唯一标识，其实就是把当前节点所在集群的名称、节点名称、节点ID等等信息封装到一起了
        this.scheduledExecutorService = ThreadUtils.newScheduledThreadPool(1,
                new ThreadFactoryImpl("BrokerControllerScheduledThread", true, getBrokerIdentity()));

        this.sendMessageExecutor = ThreadUtils.newThreadPoolExecutor(
                this.brokerConfig.getSendMessageThreadPoolNums(),
                this.brokerConfig.getSendMessageThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.sendThreadPoolQueue,
                new ThreadFactoryImpl("SendMessageThread_", getBrokerIdentity()));

        this.putMessageFutureExecutor = ThreadUtils.newThreadPoolExecutor(
                this.brokerConfig.getPutMessageFutureThreadPoolNums(),
                this.brokerConfig.getPutMessageFutureThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.putThreadPoolQueue,
                new ThreadFactoryImpl("SendMessageThread_", getBrokerIdentity()));

        //创建用来处理控制台向Broker发送的请求的线程池
        this.adminBrokerExecutor = ThreadUtils.newThreadPoolExecutor(
                this.brokerConfig.getAdminBrokerThreadPoolNums(),
                this.brokerConfig.getAdminBrokerThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.adminBrokerThreadPoolQueue,
                new ThreadFactoryImpl("AdminBrokerThread_", getBrokerIdentity()));
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/9
     * @方法描述：注册请求处理器的方法
     */
    public void registerProcessor() {
        sendMessageProcessor.registerSendMessageHook(sendMessageHookList);
        //这行代码暂时注释掉
        //sendMessageProcessor.registerConsumeMessageHook(consumeMessageHookList);
        //创建专门处理控制台请求的请求处理器
        AdminBrokerProcessor adminProcessor = new AdminBrokerProcessor(this);
        //把该处理器注册到服务器中
        this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendMessageProcessor, this.sendMessageExecutor);
        this.remotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendMessageProcessor, this.sendMessageExecutor);
        this.remotingServer.registerDefaultProcessor(adminProcessor, this.adminBrokerExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendMessageProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerProcessor(RequestCode.SEND_BATCH_MESSAGE, sendMessageProcessor, this.sendMessageExecutor);
        this.fastRemotingServer.registerDefaultProcessor(adminProcessor, this.adminBrokerExecutor);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/9
     * @方法描述：初始化Broker模块各个定时任务的方法，在第五版本代码中，这个方法的内容还很简单，随着代码版本的迭代，这个方法的内容会越来越多
     */
    protected void initializeScheduledTasks() {

        //初始化Broker模块中要执行的定时任务
        initializeBrokerScheduledTasks();

        //判断Broker配置信息中是否设置了NameServer的地址，因为Namesrv也可以构建集群，所以这里用户定义的可能是多个NameServer地址，并且用逗号分隔
        if (this.brokerConfig.getNamesrvAddr() != null) {
            //如果设置了NameServer地址，则把用户设置得NameServer地址更新到NettyRemotingClient客户端的namesrvAddrList成员变量中
            this.updateNamesrvAddr();
            LOG.info("Set user specified name server address: {}", this.brokerConfig.getNamesrvAddr());
            //提交一个定期更新NameServer地址的任务
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    try {
                        //在这里执行更新NameServer地址的操作
                        //之所以要定期更新NameServer地址，是因为NameServer地址可能会发生变化，也许构建的是Namesrv集群
                        //集群中会添加新的节点，如果有新的Namesrv添加到集群中了，就要让Broker知道，所以需要定期更新Broker中的NameServer地址
                        BrokerController.this.updateNamesrvAddr();
                    } catch (Throwable e) {
                        LOG.error("Failed to update nameServer address list", e);
                    }
                }
            }, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
            //执行到这里意味着用户没有在配置文件中设置Namesrv地址，而是把获取Namesrv地址的任务交给了地址服务器来保存
            //当前Broker客户端需要定期访问地址服务器，获取Namesrv的最新地址
        } else if (this.brokerConfig.isFetchNamesrvAddrByAddressServer()) {
            //这个定时任务就会让当前Broker客户端定期访问地址服务器，获取Namesrv的最新地址
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    try {
                        BrokerController.this.brokerOuterAPI.fetchNameServerAddr();
                    } catch (Throwable e) {
                        LOG.error("Failed to fetch nameServer address", e);
                    }
                }
            }, 1000 * 10, this.brokerConfig.getFetchNamesrvAddrInterval(), TimeUnit.MILLISECONDS);
        }
    }



    //这个定时任务会定期打印Broker中使用的各个执行器的水位信息，用于监控内部状态和性能指标
    //在第五版本代码中，这个方法的内容还很简单，随着代码版本的迭代，这个方法的内容会越来越多
    protected void initializeBrokerScheduledTasks() {

        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    BrokerController.this.printWaterMark();
                } catch (Throwable e) {
                    LOG.error("BrokerController: failed to print broker watermark", e);
                }
            }
        }, 10, 1, TimeUnit.SECONDS);

    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/9
     * @方法描述：更新NameServer地址的方法
     */
    private void updateNamesrvAddr() {
        //判断是否通过DNS查找Namesrv地址
        if (this.brokerConfig.isFetchNameSrvAddrByDnsLookup()) {
            this.brokerOuterAPI.updateNameServerAddressListByDnsLookup(this.brokerConfig.getNamesrvAddr());
        } else {
            //直接使用用户配置的Namesrv地址更新Namesrv地址列表
            this.brokerOuterAPI.updateNameServerAddressList(this.brokerConfig.getNamesrvAddr());
        }
    }


    //初始化RPC回调对象的方法
    private void initialRpcHooks() {
        List<RPCHook> rpcHooks = ServiceProvider.load(RPCHook.class);
        if (rpcHooks == null || rpcHooks.isEmpty()) {
            return;
        }
        for (RPCHook rpcHook : rpcHooks) {
            this.registerServerRPCHook(rpcHook);
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/9
     * @方法描述：启动Broker开始工作的方法
     */
    public void start() throws Exception {

        //启动Broker组件对外发送请求的API组件
        if (this.brokerOuterAPI != null) {
            this.brokerOuterAPI.start();
        }
        //启动Broker模块的基础服务
        startBasicService();

        //这里执行了一个很重要的操作，也是第五版本代码着重实现的操作，那就是把Broker的信息注册到Namesrv中
        this.registerBrokerAll(true, false, true);

        //接下来又创建并提交了一个定时任务，这个定时任务会定期把当前Broker的信息注册到Namesrv中
        scheduledFutures.add(this.scheduledExecutorService.scheduleAtFixedRate(new AbstractBrokerRunnable(this.getBrokerIdentity()) {

            @Override
            public void run0() {
                try {
                    //如果Broker处于隔离状态，则不需要注册信息到Namesrv中
                    if (isIsolated) {
                        BrokerController.LOG.info("Skip register for broker is isolated");
                        return;
                    }
                    //执行向所有Namesrv节点注册当前Broker信息的操作
                    BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
                } catch (Throwable e) {
                    BrokerController.LOG.error("registerBrokerAll Exception", e);
                }
            }
        }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS));

    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/9
     * @方法描述：启动Broker模块基础服务的方法
     */
    protected void startBasicService() throws Exception {

        //启动默认消息存储引擎
        if (this.messageStore != null) {
            this.messageStore.start();
        }

        //启动Netty服务端
        if (this.remotingServer != null) {
            this.remotingServer.start();
            //设置Netty服务端的监听端口
            if (null != nettyServerConfig && 0 == nettyServerConfig.getListenPort()) {
                nettyServerConfig.setListenPort(remotingServer.localListenPort());
            }
        }

        //启动快速响应Netty服务端
        if (this.fastRemotingServer != null) {
            this.fastRemotingServer.start();
        }

        this.storeHost = new InetSocketAddress(this.getBrokerConfig().getBrokerIP1(), this.getNettyServerConfig().getListenPort());

        //启动所有Broker插件
        for (BrokerAttachedPlugin brokerAttachedPlugin : brokerAttachedPlugins) {
            if (brokerAttachedPlugin != null) {
                brokerAttachedPlugin.start();
            }
        }


        //启动Broker统计管理器
        if (this.brokerStatsManager != null) {
            this.brokerStatsManager.start();
        }

        //启动客户端连接清理器
        if (this.clientHousekeepingService != null) {
            this.clientHousekeepingService.start();
        }

    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/9
     * @方法描述：把当前Broker自身信息注册到所有Namesrv节点的方法
     */
    public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway, boolean forceRegister) {
        //得到当前Broker的主题配置信息
        ConcurrentMap<String, TopicConfig> topicConfigMap = this.getTopicConfigManager().getTopicConfigTable();
        ConcurrentHashMap<String, TopicConfig> topicConfigTable = new ConcurrentHashMap<>();
        //遍历主题配置信息
        for (TopicConfig topicConfig : topicConfigMap.values()) {
            //判断当前Broker是否有权限读写该主题
            if (!PermName.isWriteable(this.getBrokerConfig().getBrokerPermission()) || !PermName.isReadable(this.getBrokerConfig().getBrokerPermission())) {
                //如果读写权限受限，则需要将该主题配置信息的权限与Broker权限进行与运算
                topicConfigTable.put(topicConfig.getTopicName(),
                        new TopicConfig(topicConfig.getTopicName(),
                                topicConfig.getReadQueueNums(),
                                topicConfig.getWriteQueueNums(),
                                //这里就是把主题配置信息的权限与Broker权限进行与运算，得到该主题真正的权限
                                topicConfig.getPerm() & getBrokerConfig().getBrokerPermission()));
            } else {
                //如果读写权限没有受限，则直接使用主题配置信息
                topicConfigTable.put(topicConfig.getTopicName(), topicConfig);
            }
            //判断当前Broker是否启用了分片注册，所谓分片注册就是在主题配置信息非常多的情况下，把主题配置信息分批发送给Namesrv
            //topicConfigTable.size() >= this.brokerConfig.getSplitRegistrationSize()这个条件判断就是用来判断是否需还需要接着分片注册主题配置信息到Namesrv
            //不需要的话就可以继续之后该分支之后的代码了
            if (this.brokerConfig.isEnableSplitRegistration() && topicConfigTable.size() >= this.brokerConfig.getSplitRegistrationSize()) {
                //构建主题配置序列化包装器
                TopicConfigAndMappingSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildSerializeWrapper(topicConfigTable);
                //把当前Broker的主题配置信息注册到所有Namesrv节点
                doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);
                //清空主题配置信息，准备下一批次的注册
                topicConfigTable.clear();
            }
        }

        //下面着段代码目前还用不到，大家直接忽略即可
        Map<String, TopicQueueMappingInfo> topicQueueMappingInfoMap = this.getTopicQueueMappingManager().getTopicQueueMappingTable().entrySet().stream()
                .map(entry -> new AbstractMap.SimpleImmutableEntry<>(entry.getKey(), TopicQueueMappingDetail.cloneAsMappingInfo(entry.getValue())))
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

        //在这里又TopicConfigAndMappingSerializeWrapper对象，这个对象包含了主题配置信息和主题队列映射信息，最终发送给Namesrv的就是这个对象
        TopicConfigAndMappingSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildSerializeWrapper(topicConfigTable, topicQueueMappingInfoMap);

        //判断是否需要注册Broker信息，这里的needRegister()方法是用来判断当前Broker是否需要向Namesrv重新注册
        //当然，因为forceRegister参数为true，所以这里肯定会执行向Namesrv注册Broker信息的操作
        if (this.brokerConfig.isEnableSplitRegistration() || forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),
                this.getBrokerAddr(),
                this.brokerConfig.getBrokerName(),
                this.brokerConfig.getBrokerId(),
                this.brokerConfig.getRegisterBrokerTimeoutMills(),
                this.brokerConfig.isInBrokerContainer())) {
            //执行注册Broker信息的操作
            doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/19
     * @方法描述：判断是否需要向Namesrv重新注册Broker信息的方法
     */
    private boolean needRegister(final String clusterName,
                                 final String brokerAddr,
                                 final String brokerName,
                                 final long brokerId,
                                 final int timeoutMills,
                                 final boolean isInBrokerContainer) {
        TopicConfigSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildTopicConfigSerializeWrapper();
        //向Namesrv发送查询数据版本的请求，如果Namesrv的数据版本和Broker的数据版本不一致，那就需要重新注册Broker信息
        List<Boolean> changeList = brokerOuterAPI.needRegister(clusterName, brokerAddr, brokerName, brokerId, topicConfigWrapper, timeoutMills, isInBrokerContainer);
        boolean needRegister = false;
        //有一个Namesrv的数据版本和Broker的数据版本不一致，就需要重新注册Broker信息
        for (Boolean changed : changeList) {
            if (changed) {
                needRegister = true;
                break;
            }
        }
        return needRegister;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/9
     * @方法描述：把当前Broker自身信息注册到所有NameServer节点的方法
     */
    protected void doRegisterBrokerAll(boolean checkOrderConfig, boolean oneway,TopicConfigSerializeWrapper topicConfigWrapper) {

        //判断当前Broker是否已经关闭，如果已经关闭，则不需要再注册Broker信息
        if (shutdown) {
            BrokerController.LOG.info("BrokerController#doRegisterBrokerAll: broker has shutdown, no need to register any more.");
            return;
        }

        //调用向外部发送请求的API组件的registerBrokerAll()方法，向所有Namesrv注册Broker信息，并且得到服务端返回的注册结果
        List<RegisterBrokerResult> registerBrokerResultList = this.brokerOuterAPI.registerBrokerAll(
                //注册当前Broker节点所在集群的名称
                this.brokerConfig.getBrokerClusterName(),
                //注册当前Broker节点的网络地址
                this.getBrokerAddr(),
                //注册当前Broker节点的名称
                this.brokerConfig.getBrokerName(),
                //注册当前Broker节点的ID，Broker集群中主节点的Id为0，从节点的Id都>0
                this.brokerConfig.getBrokerId(),
                //注册Broker节点的高可用服务地址
                //这里我简单解释一下，虽然还没有引入Broker集群，但是简单解释一下概念是没问题的
                //所谓高可用地址其实就是Broker从节点要从主节点备份数据的地址，假如说当前节点是主节点
                //并且当前节点有一个HAPort，也就是高可用地址中的端口号，那么从节点就可以连接到主节点的HAPort
                //从这个地址备份主节点的消息数据。这里有一个问题，假如当前Broker节点不是主节点，那把它的高可用地址注册到Namesrv有什么用呢？
                //大家可以先记住这个问题，后面引入集群后，我再来为大家解答这个问题
                this.getHAServerAddr(),
                topicConfigWrapper,
                //这个方法参数暂时用不上，大家可以暂时忽略
                Lists.newArrayList(),
                //表示发送请求是否为单项请求
                oneway,
                //注册Broker节点信息的超时时间
                this.brokerConfig.getRegisterBrokerTimeoutMills(),
                //是否启用了acting-master模式，这个也是Broker集群模式下的一个概念，就是主节点宕机的时候
                //从节点可以代理主节点，并且代理的权限扩大了，现在大家可以先不必关注这个概念
                this.brokerConfig.isEnableSlaveActingMaster(),
                //是否启用压缩注册，其实就是把请求体的内容压缩一下，在第五版本代码还用不到这个参数，其实在源码中也不会用到这个功能
                //因为源码中默认压缩功能是不开启的
                this.brokerConfig.isCompressedRegister(),
                //如果启用了acting-master模式，那就传递Broker的心跳超时时间
                this.brokerConfig.isEnableSlaveActingMaster() ? this.brokerConfig.getBrokerNotActiveTimeoutMillis() : null,
                //注册当前Broker节点的唯一标识
                this.getBrokerIdentity());

        //处理注册结果
        handleRegisterBrokerResult(registerBrokerResultList, checkOrderConfig);
    }


    //处理Broker节点向Namesrv注册结果的方法，在第七版本代码中，该方法仍然是一个伪实现，引入集群之后再来完善这个方法
    protected void handleRegisterBrokerResult(List<RegisterBrokerResult> registerBrokerResultList, boolean checkOrderConfig) {
        for (RegisterBrokerResult registerBrokerResult : registerBrokerResultList) {
            System.out.println("接收到了来自NameServer的注册结果：" + registerBrokerResult);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/9
     * @方法描述：关闭Broker程序的方法
     */
    public void shutdown() {
        //关闭内部所有基础服务组件
        shutdownBasicService();

        //取消所有已提交的定时任务
        for (ScheduledFuture<?> scheduledFuture : scheduledFutures) {
            scheduledFuture.cancel(true);
        }

        //关闭对外发送请求的API组件
        if (this.brokerOuterAPI != null) {
            this.brokerOuterAPI.shutdown();
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/9
     * @方法描述：关闭Broker内部所有基础服务组件的方法
     */
    protected void shutdownBasicService() {
        //更新关闭标志为true
        shutdown = true;

        //注销注册到Namesrv上的Broker信息
        this.unregisterBrokerAll();

        if (this.shutdownHook != null) {
            this.shutdownHook.beforeShutdown(this);
        }
        if (this.remotingServer != null) {
            this.remotingServer.shutdown();
        }
        if (this.fastRemotingServer != null) {
            this.fastRemotingServer.shutdown();
        }
        if (this.brokerStatsManager != null) {
            this.brokerStatsManager.shutdown();
        }
        if (this.clientHousekeepingService != null) {
            this.clientHousekeepingService.shutdown();
        }
        if (this.messageStore != null) {
            this.messageStore.shutdown();
        }
        shutdownScheduledExecutorService(this.scheduledExecutorService);
        if (this.sendMessageExecutor != null) {
            this.sendMessageExecutor.shutdown();
        }

        if (this.putMessageFutureExecutor != null) {
            this.putMessageFutureExecutor.shutdown();
        }

        if (this.adminBrokerExecutor != null) {
            this.adminBrokerExecutor.shutdown();
        }
        for (BrokerAttachedPlugin brokerAttachedPlugin : brokerAttachedPlugins) {
            if (brokerAttachedPlugin != null) {
                brokerAttachedPlugin.shutdown();
            }
        }
    }


    //终止定时任务执行器停止工作的方法
    protected void shutdownScheduledExecutorService(ScheduledExecutorService scheduledExecutorService) {
        if (scheduledExecutorService == null) {
            return;
        }
        scheduledExecutorService.shutdown();
        try {
            //最多等待5秒钟，如果5秒钟内还没有关闭，则强制关闭
            scheduledExecutorService.awaitTermination(5000, TimeUnit.MILLISECONDS);
        } catch (InterruptedException ignore) {
            BrokerController.LOG.warn("shutdown ScheduledExecutorService was Interrupted!  ", ignore);
            Thread.currentThread().interrupt();
        }
    }


    //打印各个执行器队列的水位信息的方法，在第五版本代码中，这个方法的内容非常简单，随着代码版本的迭代，这个方法的内容会越来越多
    public void printWaterMark() {
        LOG_WATER_MARK.info("[WATERMARK] Send Queue Size: {} SlowTimeMills: {}", this.sendThreadPoolQueue.size(), headSlowTimeMills4SendThreadPoolQueue());
        LOG_WATER_MARK.info("[WATERMARK] Admin Queue Size: {} SlowTimeMills: {}", this.adminBrokerThreadPoolQueue.size(), headSlowTimeMills(this.adminBrokerThreadPoolQueue));
    }

    public long headSlowTimeMills4SendThreadPoolQueue() {
        return this.headSlowTimeMills(this.sendThreadPoolQueue);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/9
     * @方法描述：计算队列头部任务慢速时间的方法
     */
    public long headSlowTimeMills(BlockingQueue<Runnable> q) {
        long slowTimeMills = 0;
        final Runnable peek = q.peek();
        if (peek != null) {
            RequestTask rt = BrokerFastFailure.castRunnable(peek);
            slowTimeMills = rt == null ? 0 : System.currentTimeMillis() - rt.getCreateTimestamp();
        }
        if (slowTimeMills < 0) {
            slowTimeMills = 0;
        }
        return slowTimeMills;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/9
     * @方法描述：注销当前Broker节点在Namesrv节点中信息的方法
     */
    protected void unregisterBrokerAll() {
        this.brokerOuterAPI.unregisterBrokerAll(
                this.brokerConfig.getBrokerClusterName(),
                this.getBrokerAddr(),
                this.brokerConfig.getBrokerName(),
                this.brokerConfig.getBrokerId());
    }


    //一下都是简单的get、set方法，我就不添加详细注释了

    public BrokerConfig getBrokerConfig() {
        return brokerConfig;
    }

    public NettyServerConfig getNettyServerConfig() {
        return nettyServerConfig;
    }

    public NettyClientConfig getNettyClientConfig() {
        return nettyClientConfig;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/9
     * @方法描述：得到当前Broker节点的完整地址的方法
     */
    public String getBrokerAddr() {
        //this.brokerConfig.getBrokerIP1()得到的就是当前Broker节点的IP地址
        //this.nettyServerConfig.getListenPort()得到的就是当前Broker节点监听的端口号
        //这里我多说一句，在brokerConfig中有一个brokerIP2成员变量，在获取Broker节点的高可用地址的时候，使用的就是这个brokerIP2
        return this.brokerConfig.getBrokerIP1() + ":" + this.nettyServerConfig.getListenPort();
    }


    //得到当前Broker节点唯一标识的方法
    public BrokerIdentity getBrokerIdentity() {

        return new BrokerIdentity(
                brokerConfig.getBrokerClusterName(),
                brokerConfig.getBrokerName(),
                brokerConfig.getBrokerId(),
                //当前Broker节点是否处于容器中运行
                brokerConfig.isInBrokerContainer());

    }


    public void registerServerRPCHook(RPCHook rpcHook) {
        getRemotingServer().registerRPCHook(rpcHook);
        this.fastRemotingServer.registerRPCHook(rpcHook);
    }


    public RemotingServer getRemotingServer() {
        return remotingServer;
    }

    public void setRemotingServer(RemotingServer remotingServer) {
        this.remotingServer = remotingServer;
    }

    public void registerClientRPCHook(RPCHook rpcHook) {
        this.getBrokerOuterAPI().registerRPCHook(rpcHook);
    }

    public BrokerOuterAPI getBrokerOuterAPI() {
        return brokerOuterAPI;
    }

    public Configuration getConfiguration() {
        return this.configuration;
    }

    public int getListenPort() {
        return this.nettyServerConfig.getListenPort();
    }

    public BrokerStatsManager getBrokerStatsManager() {
        return brokerStatsManager;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/9
     * @方法描述：得到当前Broker节点的高可用地址的方法
     */
    public String getHAServerAddr() {
        //这里使用的IP就是brokerConfig中的brokerIP2，端口号就是Netty服务端监听的端口号+1
        return this.brokerConfig.getBrokerIP2() + ":" + nettyServerConfig.getListenPort() + 1;
    }

    public TopicConfigManager getTopicConfigManager() {
        return topicConfigManager;
    }

    public void setTopicConfigManager(TopicConfigManager topicConfigManager) {
        this.topicConfigManager = topicConfigManager;
    }

    public TopicQueueMappingManager getTopicQueueMappingManager() {
        return topicQueueMappingManager;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/19
     * @方法描述：注册单个主题信息到所有Namesrv的方法
     */
    public synchronized void registerSingleTopicAll(final TopicConfig topicConfig) {
        TopicConfig tmpTopic = topicConfig;
        //判断当前Broker是否有权限读写该主题
        if (!PermName.isWriteable(this.getBrokerConfig().getBrokerPermission())
                || !PermName.isReadable(this.getBrokerConfig().getBrokerPermission())) {
            //如果读写权限受限，那么需要将该主题配置信息的权限与Broker权限进行与运算，创建一个新的主题配置信息对象
            tmpTopic = new TopicConfig(topicConfig);
            //将主题的权限与Broker的权限进行与运算，确保主题权限不超过Broker权限
            tmpTopic.setPerm(topicConfig.getPerm() & this.brokerConfig.getBrokerPermission());
        }
        //执行把单个主题信息注册到所有Namesrv的操作
        this.brokerOuterAPI.registerSingleTopicAll(this.brokerConfig.getBrokerName(), tmpTopic, 3000);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/19
     * @方法描述：将Broker内部多个主题配置信息注册到所有Namesrv的方法，该方法和以上方法的区别是，上面的方法只会把单个主题信息注册到所有Namesrv，而这个方法会把Broker内部存储的多个主题信息注册到所有Namesrv
     */
    public synchronized void registerIncrementBrokerData(TopicConfig topicConfig, DataVersion dataVersion) {
        this.registerIncrementBrokerData(Collections.singletonList(topicConfig), dataVersion);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/19
     * @方法描述：将Broker内部多个主题配置信息注册到所有Namesrv的方法
     */
    public synchronized void registerIncrementBrokerData(List<TopicConfig> topicConfigList, DataVersion dataVersion) {
        if (topicConfigList == null || topicConfigList.isEmpty()) {
            return;
        }
        TopicConfigAndMappingSerializeWrapper topicConfigSerializeWrapper = new TopicConfigAndMappingSerializeWrapper();
        //把Broker内部存储的主题配置信息的数据版本设置到序列化包装器中
        topicConfigSerializeWrapper.setDataVersion(dataVersion);

        //收集Broker内部存储的所有主题配置信息，并且根据Broker的读写权限得到真正的主题配置信息
        //下面这些操作在上面的方法中见过很多次了，我就不添加详细注释了
        ConcurrentMap<String, TopicConfig> topicConfigTable = topicConfigList.stream()
                .map(topicConfig -> {
                    TopicConfig registerTopicConfig;
                    if (!PermName.isWriteable(this.getBrokerConfig().getBrokerPermission())
                            || !PermName.isReadable(this.getBrokerConfig().getBrokerPermission())) {
                        registerTopicConfig =
                                new TopicConfig(topicConfig.getTopicName(),
                                        topicConfig.getReadQueueNums(),
                                        topicConfig.getWriteQueueNums(),
                                        topicConfig.getPerm() & this.brokerConfig.getBrokerPermission(),
                                        topicConfig.getTopicSysFlag());
                    } else {
                        registerTopicConfig = new TopicConfig(topicConfig);
                    }
                    return registerTopicConfig;
                }).collect(Collectors.toConcurrentMap(TopicConfig::getTopicName, Function.identity()));

        //把收集到的主题配置信息设置到序列化包装器中
        topicConfigSerializeWrapper.setTopicConfigTable(topicConfigTable);

        //以下这段代码还用不到，大家直接忽略即可
        Map<String, TopicQueueMappingInfo> topicQueueMappingInfoMap = topicConfigList.stream()
                .map(TopicConfig::getTopicName)
                .map(topicName -> Optional.ofNullable(this.topicQueueMappingManager.getTopicQueueMapping(topicName))
                        .map(info -> new AbstractMap.SimpleImmutableEntry<>(topicName, TopicQueueMappingDetail.cloneAsMappingInfo(info)))
                        .orElse(null))
                .filter(Objects::nonNull)
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
        if (!topicQueueMappingInfoMap.isEmpty()) {
            topicConfigSerializeWrapper.setTopicQueueMappingInfoMap(topicQueueMappingInfoMap);
        }


        //执行把Broker内部存储的多个主题信息注册到所有Namesrv的操作
        doRegisterBrokerAll(true, false, topicConfigSerializeWrapper);
    }

    public ProducerManager getProducerManager() {
        return producerManager;
    }

    public ConsumerManager getConsumerManager() {
        return consumerManager;
    }

    public NettyRequestProcessor getSendMessageProcessor() {
        return sendMessageProcessor;
    }

    public List<SendMessageHook> getSendMessageHookList() {
        return sendMessageHookList;
    }

    public void setStoreHost(InetSocketAddress storeHost) {
        this.storeHost = storeHost;
    }

    public InetSocketAddress getStoreHost() {
        return storeHost;
    }

    public void registerSendMessageHook(final SendMessageHook hook) {
        this.sendMessageHookList.add(hook);
        LOG.info("register SendMessageHook Hook, {}", hook.hookName());
    }

    public BlockingQueue<Runnable> getSendThreadPoolQueue() {
        return sendThreadPoolQueue;
    }

    public MessageStore getMessageStore() {
        return messageStore;
    }

    public MessageStoreConfig getMessageStoreConfig() {
        return messageStoreConfig;
    }

    public Executor getPutMessageFutureExecutor() {
        return putMessageFutureExecutor;
    }

    public BlockingQueue<Runnable> getAsyncPutThreadPoolQueue() {
        return putThreadPoolQueue;
    }
}
